package SparkStreaming

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}

import scala.collection.mutable.{ArrayBuffer, Map, Set}

object SparkStreaming03 {
  def main(args: Array[String]): Unit = {

   /* val brokers=PropertieUtil.getString("kafka.brokers")
    val topic=PropertieUtil.getString("kafka.topic")
    val tableNameHn=PropertieUtil.getString(("hbase.tablenamehn"))
    val tableNamesd=PropertieUtil.getString(("hbase.tablenamesd"))
    val hbaseZookeeperList=PropertieUtil.getString("hbase.zookeeper")*/

    val sparkconf=new SparkConf().setMaster("local[2]").setAppName("SparkStreaming")

    val ssc=new StreamingContext(sparkconf,Seconds(10))
    val accumlators=ssc.sparkContext.accumulator(0)
    val topics=Set("topc")
    val kafkaParams=Map[String,Object](
      "bootstrap.servers" ->"brokers",
      "key.deserializer" ->classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "*g",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )
    var kafkaStream=KafkaUtils.createDirectStream[String,String](ssc, PreferConsistent,Subscribe[String,String](topics,kafkaParams))
    var events=kafkaStream.map(record=>record.value())
    events.foreachRDD(rdd=>{
      val sqlContext = SQLContext.getOrCreate(rdd.sparkContext)
      import sqlContext.implicits._
      rdd.map(pair=>(dataArray(pair))).map(v=>(v(0),v(1),v(2))).toDF(colNames = "PROID","KEY","VALUE").registerTempTable("speedtable")
      val dataframe=sqlContext.sql("select * from  speedtable ")
      dataframe.show()
      println(dataframe.count())
    })

    ssc.start()
    ssc.awaitTermination()
  }

  def dataArray(pair:String):ArrayBuffer[String]={
    val array = ArrayBuffer[String]()
/*    val JsonObject=JSONObject.fromObject(pair)
    val pro=JsonObject.get("A").toString
    array.append(pro)
    val key=JsonObject.get("B").toString
    array.append(key)
    val data=JsonObject.get("C").toString
    array.append(data)*/
    array
  }
}